#include "BBTCP.h"
#include <mstcpip.h>

#pragma comment(lib, "ws2_32.lib")
#pragma comment(lib, "mswsock.lib")

IOCPService::IOCPService()
{
    m_hIOCP = INVALID_HANDLE_VALUE;
    m_stop = false;
    m_threads.clear();
    m_port = 0;
    m_listenSocket = INVALID_SOCKET;
    m_listenSocket_IPv6 = INVALID_SOCKET;
    m_ipv4Available = false;
    m_ipv6Available = false;
    m_pPFEventHandler = NULL;
    m_ovDataCounter = 0;
    m_pAcceptEx = NULL;
    m_pConnectEx = NULL;
    m_pGetAcceptExSockaddrs = NULL;
    InitializeListHead(&m_ovDataList);
}

IOCPService::~IOCPService()
{
    destory();
}
STATUS IOCPService::failExit()
{
    destory();
    return STATUS_FAIL;
}
void IOCPService::destory()
{
    m_stop = true;
    m_port = 0;
    {
        std::lock_guard<std::mutex> lock{m_taskSyncLock};
        cv_task.notify_all();
    }
    if (m_hIOCP != INVALID_HANDLE_VALUE)
    {
        ::CloseHandle(m_hIOCP);
        m_hIOCP = INVALID_HANDLE_VALUE;
    }
    if (m_listenSocket != INVALID_SOCKET)
    {
        ::closesocket(m_listenSocket);
        m_listenSocket = INVALID_SOCKET;
    }

    if (m_listenSocket_IPv6 != INVALID_SOCKET)
    {
        ::closesocket(m_listenSocket_IPv6);
        m_listenSocket_IPv6 = INVALID_SOCKET;
    }

    for (auto &iter : m_threads)
    {
        if (iter->joinable())
            iter->join();
        free(iter);
    }

    m_threads.clear();

    {
        std::lock_guard<std::mutex> lock{m_ConnectLock};
        for (auto &iter : m_connectlist)
        {
            ::shutdown(iter.first, SD_BOTH);
        }
        m_connectlist.clear();
    }
    ::WSACleanup();
    m_ipv4Available = false;
    m_ipv6Available = false;
    std::lock_guard<std::mutex> lock{ m_ovDataLock };
    while (!IsListEmpty(&m_ovDataList))
    {
        PLIST_ENTRY pEntry = RemoveTailList(&m_ovDataList);
        POV pov = CONTAINING_RECORD(pEntry, OV_DATA, entry);
        DbgPrint("Free id=%ld,type=%ld, socket=%ld", pov->id, pov->type, pov->socket);
        delete pov;
    }
}
int IOCPService::getPort()
{
    return m_port;
}

bool IOCPService::isIPFamilyAvailable(int ipFamily)
{
    switch (ipFamily)
    {
    case AF_INET:
        return m_ipv4Available;
    case AF_INET6:
        return m_ipv6Available;
    }
    return false;
}
STATUS IOCPService::init(int mode, int port)
{
    if (INVALID_HANDLE_VALUE != m_hIOCP) return STATUS_SUCCESS;

    WSAData wsa;
    int optval = 1;

    if (::WSAStartup(MAKEWORD(2, 2), &wsa) != 0) return failExit();

    if (LOBYTE(wsa.wVersion) != 2 || HIBYTE(wsa.wVersion) != 2) return failExit();
    if (!initExtensions()) return failExit();

    m_hIOCP = CreateIoCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);
    if (NULL == m_hIOCP) return failExit();
    std::thread *task = new std::thread([this]
                                        { this->workerThread(); });
    if (!task) return failExit();
    task->detach();
    m_threads.push_back(task);

    SYSTEM_INFO info;
    ::GetSystemInfo(&info);

    int threadCount = info.dwNumberOfProcessors;
    if (threadCount <= 1) threadCount = 2;

    m_port = port;

    if ((mode & IOCP_SERVICE_IPV4) == IOCP_SERVICE_IPV4)
    {
        sockaddr_in addr;
        memset(&addr, 0, sizeof(addr));
        addr.sin_family = AF_INET;
        addr.sin_port = htons(m_port);

        m_listenSocket = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
        if (m_listenSocket == INVALID_SOCKET) return STATUS_FAIL;

        if (setsockopt(m_listenSocket, SOL_SOCKET,SO_REUSEADDR,(char *)&optval, sizeof(optval)) != 0) return failExit();

        if (bind(m_listenSocket, (SOCKADDR*)&addr, sizeof(addr)) != 0) return failExit();

        if (listen(m_listenSocket, SOMAXCONN) != 0) return failExit();

        registerSocket(m_listenSocket);

        if (!startAccept(m_listenSocket)) return failExit();
        m_ipv4Available = true;
        threadCount += 1;
    }
    if ((mode & IOCP_SERVICE_IPV6) == IOCP_SERVICE_IPV6)
    {
        sockaddr_in6 addr;

        memset(&addr, 0, sizeof(addr));
        addr.sin6_family = AF_INET6;
        addr.sin6_port = htons(m_port);

        m_listenSocket_IPv6 = WSASocket(AF_INET6, SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
        if (m_listenSocket_IPv6 == INVALID_SOCKET) return failExit();

        if (setsockopt(m_listenSocket_IPv6, SOL_SOCKET, SO_REUSEADDR, (char*)&optval, sizeof(optval)) != 0) return failExit();

        if (bind(m_listenSocket_IPv6, (SOCKADDR*)&addr, sizeof(addr)) != 0) return failExit();

        if (listen(m_listenSocket_IPv6, SOMAXCONN) != 0) return failExit();

        registerSocket(m_listenSocket_IPv6);

        if (!startAccept(m_listenSocket_IPv6)) return failExit();

        m_ipv6Available = true;
        threadCount += 1;
    }

    /*
     * If onAcceptComplete() fails to add a new accept, it will occupy
     * the task for a long time and allocate an additional 1 or 2 threads
     * threadCount has increased
     */

    for (int i = 0; i < threadCount; i++)
    {
        std::thread* task = new std::thread([this]
            { this->execute(); });

        if (!task) return failExit();
        task->detach();
        m_threads.push_back(task);
    }
    return STATUS_SUCCESS;
}
void IOCPService::addConnect(ENDPOINT_ID eID, LPVOID pData)
{
    std::lock_guard<std::mutex> lock{m_ConnectLock};
    m_connectlist[eID] = pData;
}

bool IOCPService::setPrivate(ENDPOINT_ID eID, LPVOID pData)
{
    std::lock_guard<std::mutex> lock{ m_ConnectLock };
    
    if (m_connectlist.find(eID) != m_connectlist.end())
    {
        m_connectlist[eID] = pData;
        return TRUE;
    }
    return FALSE;
}

LPVOID IOCPService::getPrivate(ENDPOINT_ID eID)
{
    std::lock_guard<std::mutex> lock{ m_ConnectLock };

    auto iter = m_connectlist.find(eID);
    if (iter != m_connectlist.end())
    {
        return iter->second;
    }
    return NULL;
}

bool IOCPService::startAccept(ENDPOINT_ID id)
{

    SOCKET s = WSASocket(((m_listenSocket == id) ? AF_INET : AF_INET6), SOCK_STREAM, IPPROTO_TCP, NULL, 0, WSA_FLAG_OVERLAPPED);
    if (s == INVALID_SOCKET)
    {
        DbgPrint("invalid id=%ld, socket=%ld", id, s);
        return false;
    }
    DbgPrint("id=%ld, socket=%ld", id, s);

    OV_DATA *pov = newOV_DATA();
    DWORD dwBytes;

    pov->type = OVT_ACCEPT;
    pov->id = id;
    pov->socket = s;
    pov->packet.Set(NULL, 2 * (sizeof(sockaddr_in6) + 16));

    if (!::AcceptEx(id,
                    s,
                    pov->packet().buf,
                    0,
                    sizeof(sockaddr_in6) + 16,
                    sizeof(sockaddr_in6) + 16,
                    &dwBytes,
                    &pov->olp))
    {
        if (WSAGetLastError() != ERROR_IO_PENDING)
        {
            DbgPrint("AcceptEx id=%ld, socket=%ld", pov->id, pov->socket);
            startClose(s);
            deleteOV_DATA (pov);
            return false;
        }
    }

    return true;
}
bool IOCPService::initExtensions()
{
    const GUID acceptex = WSAID_ACCEPTEX;
    const GUID connectex = WSAID_CONNECTEX;
    const GUID getacceptexsockaddrs = WSAID_GETACCEPTEXSOCKADDRS;

    SOCKET s = socket(AF_INET, SOCK_STREAM, 0);
    if (s == INVALID_SOCKET)
        return false;

    m_pAcceptEx = (LPFN_ACCEPTEX)getExtensionFunction(s, &acceptex);
    m_pConnectEx = (LPFN_CONNECTEX)getExtensionFunction(s, &connectex);
    m_pGetAcceptExSockaddrs = (LPFN_GETACCEPTEXSOCKADDRS)getExtensionFunction(s, &getacceptexsockaddrs);

    closesocket(s);

    return m_pAcceptEx != NULL && m_pConnectEx != NULL && m_pGetAcceptExSockaddrs != NULL;
}
void *IOCPService::getExtensionFunction(SOCKET s, const GUID *which_fn)
{
    void *ptr = NULL;
    DWORD bytes = 0;
    WSAIoctl(s,
             SIO_GET_EXTENSION_FUNCTION_POINTER,
             (GUID *)which_fn, sizeof(*which_fn),
             &ptr, sizeof(ptr),
             &bytes,
             NULL,
             NULL);
    return ptr;
}
bool IOCPService::startConnect(SOCKET socket, sockaddr *pAddr, int addrLen, ENDPOINT_ID id)
{
    OV_DATA *pov = newOV_DATA();
    pov->type = OVT_CONNECT;
    pov->id = id;
    pov->socket = socket;

    DbgPrint("id=%ld, socket=%ld", id, socket);

    if (pAddr->sa_family == AF_INET)
    {
        struct sockaddr_in addr;
        ZeroMemory(&addr, sizeof(addr));
        addr.sin_family = AF_INET;

        bind(socket, (SOCKADDR *)&addr, sizeof(addr));
    }
    else
    {
        struct sockaddr_in6 addr;
        ZeroMemory(&addr, sizeof(addr));
        addr.sin6_family = AF_INET6;

        bind(socket, (SOCKADDR *)&addr, sizeof(addr));
    }

    if (!m_pConnectEx(socket, pAddr, addrLen, NULL, 0, 0, &pov->olp))
    {
        int err = WSAGetLastError();
        if (err != ERROR_IO_PENDING)
        {
            DbgPrint("id=%ld, error=%ld", id, GetLastError());
            deleteOV_DATA (pov);
            return false;
        }
    }

    return true;
}

STATUS IOCPService::close(ENDPOINT_ID id)
{
    std::lock_guard<std::mutex> lock{m_ConnectLock};
    auto iter = m_connectlist.find(id);
    if (iter != m_connectlist.end())
    {
        startClose(iter->first);
        return STATUS_SUCCESS;
    }
    return STATUS_INVALID_ENDPOINT_ID;
}
STATUS IOCPService::sendMsg(ENDPOINT_ID id, const char *buf, int len)
{
    OV_DATA *pov;
    DWORD dwBytes;
    SOCKET socket = INVALID_SOCKET;
    {
        std::lock_guard<std::mutex> lock{m_ConnectLock};
        auto iter = m_connectlist.find(id);
        if (iter != m_connectlist.end())
        {
            socket = iter->first;
        }
    }
    if (socket == INVALID_SOCKET)
    {
        return STATUS_INVALID_ENDPOINT_ID;
    }

    pov = newOV_DATA();

    pov->type = OVT_SEND;
    pov->id = id;
    pov->socket = socket;
    pov->packet.Set((char *)buf, len);

    if (WSASend(pov->socket,
                &pov->packet(), 1,
                &dwBytes, 0, &pov->olp, NULL) != 0)
    {
        int err = WSAGetLastError();
        if (err != ERROR_IO_PENDING)
        {
            DbgPrint("id=%ld, error=%ld", id, GetLastError());
            deleteOV_DATA (pov);
            return STATUS_FAIL;
        }
    }

    return STATUS_SUCCESS;
}
ENDPOINT_ID IOCPService::connect(const char *szIp, int port, int family, LPVOID pData)
{
    SOCKET tcpSocket = WSASocket((family == AF_INET) ? AF_INET : AF_INET6,
                                 SOCK_STREAM, IPPROTO_TCP, 0, 0, WSA_FLAG_OVERLAPPED);

    if (tcpSocket == INVALID_SOCKET)
    {
        return STATUS_INVALID_ENDPOINT_ID;
    }
    if (!registerSocket(tcpSocket))
    {
        ::closesocket(tcpSocket);
        return STATUS_INVALID_ENDPOINT_ID;
    }

    char remoteAddress[MAX_ADDRESS_LENGTH] = {0};

    if ((family & AF_INET) == AF_INET)
    {
        PSOCKADDR_IN server_address = PSOCKADDR_IN(remoteAddress);
        server_address->sin_family = family;
        server_address->sin_addr.S_un.S_addr = inet_addr(szIp);
        server_address->sin_port = htons(port);
    }
    else
    {

        PSOCKADDR_IN6_LH server_address = (PSOCKADDR_IN6_LH)remoteAddress;
        server_address->sin6_family = AF_INET6;
        server_address->sin6_port = htons(port);
        inet_pton(AF_INET6, szIp, &server_address->sin6_addr);
    }

    addConnect(tcpSocket, pData);

    if (!startConnect(tcpSocket, (PSOCKADDR)remoteAddress, sizeof(SOCKADDR), tcpSocket))
    {
        return STATUS_INVALID_ENDPOINT_ID;
    }

    return tcpSocket;
}

void IOCPService::workerThread()
{
    while (!m_stop)
    {
        DWORD dwTransferred = 0;
        ULONG_PTR cKey = NULL;
        OVERLAPPED *pOverlapped = NULL;

        if (GetQueuedCompletionStatus(m_hIOCP, &dwTransferred, &cKey, &pOverlapped, 500))
        {
            dispatch((SOCKET)cKey, dwTransferred, pOverlapped, 0);
        }
        else
        {
            DWORD err = GetLastError();
            if (err != WAIT_TIMEOUT)
            {
                dispatch((SOCKET)cKey, dwTransferred, pOverlapped, err);
            }
        }
    }
}

void IOCPService::dispatch(SOCKET socket, DWORD dwTransferred, OVERLAPPED *pOverlapped, int error)
{
    if (pOverlapped)
    {
        POV pov = CONTAINING_RECORD(pOverlapped, OV_DATA, olp);
        pov->dwTransferred = dwTransferred;
        pov->error = error;
        //pov->socket = socket;
        std::lock_guard<std::mutex> lock{m_taskSyncLock};
        tasks.emplace(pov);
        cv_task.notify_one();
    }
}

void IOCPService::execute()
{
    while (!m_stop)
    {
        POV pov;
        {
            std::unique_lock<std::mutex> lock{this->m_taskSyncLock};

            this->cv_task.wait(lock,
                               [this]
                               {
                                   return this->m_stop.load() ||
                                          !this->tasks.empty();
                               });
            if (this->m_stop && this->tasks.empty())
            {
                return;
            }
            pov = this->tasks.front();
            this->tasks.pop();
        }

        if (pov)
        {
            switch (pov->type)
            {
            case OVT_ACCEPT:
                onAcceptComplete(pov->socket, pov->dwTransferred, pov, pov->error);
                break;
            case OVT_CONNECT:
                onConnectComplete(pov->socket, pov->dwTransferred, pov, pov->error);
                break;
            case OVT_SEND:
                onSendComplete(pov->socket, pov->dwTransferred, pov, pov->error);
                break;
            case OVT_RECEIVE:
                onReceiveComplete(pov->socket, pov->dwTransferred, pov, pov->error);
                break;
            case OVT_CLOSE:
                onClose(pov->socket, pov->dwTransferred, pov, pov->error);
                break;
            }

            deleteOV_DATA(pov);
        }
    }
}
void IOCPService::onConnectComplete(SOCKET socket, DWORD dwTransferred, OV_DATA *pov, int error)
{
    if (error != 0)
    {
        DbgPrint("id=%ld failed, err=%ld", pov->id, error);        
        startClose(socket);
        return;
    }

    DbgPrint("id=%ld, socket=%ld", pov->id, socket);

    setKeepAliveVals(socket);

    if (m_pPFEventHandler)
    {
        sockaddr_storage storage;
        socklen_t sock_len = sizeof(storage);
        TCP_CONN_INFO connInfo = {};

        if (getsockname(socket, (sockaddr *)&storage, &sock_len) != INVALID_SOCKET)
        {
            memcpy_s(connInfo.localAddress, MAX_ADDRESS_LENGTH, (sockaddr *)&storage,
                     (storage.ss_family == AF_INET) ? sizeof(SOCKADDR) : sizeof(SOCKADDR_IN6_LH));
        }
        if (getpeername(socket, (sockaddr *)&storage, &sock_len) != INVALID_SOCKET)
        {
            memcpy_s(connInfo.remoteAddress, MAX_ADDRESS_LENGTH, (sockaddr *)&storage,
                     (storage.ss_family == AF_INET) ? sizeof(SOCKADDR) : sizeof(SOCKADDR_IN6_LH));
        }
        DbgPrint("connect socket %ld, %s:%ld -> %s:%ld", socket,
            inet_ntoa(((PSOCKADDR_IN)connInfo.localAddress)->sin_addr), ntohs(((PSOCKADDR_IN)connInfo.localAddress)->sin_port),
            inet_ntoa(((PSOCKADDR_IN)connInfo.remoteAddress)->sin_addr), ntohs(((PSOCKADDR_IN)connInfo.remoteAddress)->sin_port));
        m_pPFEventHandler->onTcpConnected(this, pov->id, &connInfo);
    }
    startReceive(pov->id, socket);
}

void IOCPService::onAcceptComplete(SOCKET acceptSocket, DWORD dwTransferred, OV_DATA *pov, int error)
{
    int ipFamily = (pov->id == m_listenSocket) ? AF_INET : AF_INET6;
    bool result = false;

    if (error != 0)
    {
        DbgPrint("Accept failed, socket= %ld, err=%ld", acceptSocket, error);
        closesocket(acceptSocket);
        addAcceptWithOK(pov);
        return;
    }

    sockaddr *pLocalAddr = NULL;
    sockaddr *pRemoteAddr = NULL;
    int localAddrLen, remoteAddrLen;

    ::GetAcceptExSockaddrs(pov->packet().buf,
                           0,
                           sizeof(sockaddr_in6) + 16,
                           sizeof(sockaddr_in6) + 16,
                           &pLocalAddr,
                           &localAddrLen,
                           &pRemoteAddr,
                           &remoteAddrLen);
#if defined(_DEBUG)
    {
        char localAddr[MAX_PATH] = "";
        char remoteAddr[MAX_PATH] = "";
        DWORD dwLen;
        sockaddr *pAddr;

        pAddr = pLocalAddr;
        dwLen = sizeof(localAddr);

        WSAAddressToString((LPSOCKADDR)pAddr,
                           (pAddr->sa_family == AF_INET6) ? sizeof(sockaddr_in6) : sizeof(sockaddr_in),
                           NULL, localAddr, &dwLen);

        pAddr = pRemoteAddr;
        dwLen = sizeof(remoteAddr);

        WSAAddressToString((LPSOCKADDR)pAddr,
                           (pAddr->sa_family == AF_INET6) ? sizeof(sockaddr_in6) : sizeof(sockaddr_in),
                           NULL, remoteAddr, &dwLen);

        DbgPrint("accepted socket %ld, %s <- %s", acceptSocket, localAddr, remoteAddr);
    }
#endif

    registerSocket(acceptSocket);

    setKeepAliveVals(acceptSocket, pov->id);

    addConnect(acceptSocket);

    if (m_pPFEventHandler)
    {
        TCP_CONN_INFO connInfo = {};

        memcpy_s(connInfo.localAddress, MAX_ADDRESS_LENGTH, pLocalAddr, localAddrLen);
        memcpy_s(connInfo.remoteAddress, MAX_ADDRESS_LENGTH, pRemoteAddr, remoteAddrLen);

        m_pPFEventHandler->onTcpAccept(this, acceptSocket, &connInfo);
    }

    startReceive(acceptSocket, acceptSocket);

    addAcceptWithOK(pov);
}

void IOCPService::addAcceptWithOK(OV_DATA* pov)
{
    while (!startAccept(pov->id))
    {
        if (m_stop) return;
        DbgPrint("startAccept %ld failed, error=%ld", pov->id, GetLastError());
        std::this_thread::sleep_for(std::chrono::milliseconds(500));
    }
}

void IOCPService::setKeepAliveVals(SOCKET socket, SOCKET listenSocket)
{
    tcp_keepalive tk;
    DWORD dwRet;
    u_long ul = 1;
    BOOL val = 1;

    if (listenSocket != INVALID_SOCKET)
    {
        setsockopt(socket, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char *)&listenSocket, sizeof(m_listenSocket));
    }
    else
    {
        setsockopt(socket, SOL_SOCKET, SO_UPDATE_CONNECT_CONTEXT, NULL, 0);
    }

    ioctlsocket(socket, FIONBIO, (u_long *)&ul);
    setsockopt(socket, IPPROTO_TCP, TCP_NODELAY, (char *)&val, sizeof(val));
    setsockopt(socket, SOL_SOCKET, SO_KEEPALIVE, (char *)&val, sizeof(val));

    tk.onoff = 1;
    tk.keepalivetime = 5 * 60 * 1000;
    tk.keepaliveinterval = 1000;

    int err = WSAIoctl(socket, SIO_KEEPALIVE_VALS,
                       (LPVOID)&tk,
                       (DWORD)sizeof(tk),
                       NULL,
                       0,
                       (LPDWORD)&dwRet,
                       NULL,
                       NULL);
    if (err != 0)
    {
        DbgPrint("WSAIoctl socket=%ld, err=%ld", socket, WSAGetLastError());
    }
}
bool IOCPService::postCompletion(SOCKET s, DWORD dwTransferred, LPOVERLAPPED pol)
{
    return PostQueuedCompletionStatus(m_hIOCP, dwTransferred, (ULONG_PTR)s, pol) ? true : false;
}

bool IOCPService::registerSocket(SOCKET s)
{
    if (!CreateIoCompletionPort((HANDLE)s, m_hIOCP, (ULONG_PTR)s, 1))
        return false;

    return true;
}
void IOCPService::onSendComplete(SOCKET socket, DWORD dwTransferred, OV_DATA *pov, int error)
{
    DbgPrint("id=%ld, socket=%ld error=%ld", pov->id, socket, error);

    if (m_pPFEventHandler)
    {
        m_pPFEventHandler->onTcpCanSend(this, pov->id);
    }
}

void IOCPService::onReceiveComplete(SOCKET socket, DWORD dwTransferred, OV_DATA *pov, int error)
{
    DbgPrint("id=%ld socket=%ld, bytes=%ld", pov->id, socket, dwTransferred);
    if (dwTransferred == 0)
    {
        if (error != 0)
        {
            DbgPrint("id=%ld, error=%ld", pov->id, error);
        }
        startClose(socket);
        return;
    }
    if (m_pPFEventHandler)
    {
        m_pPFEventHandler->onTcpReceive(this, pov->id, &pov->packet());
    }
    startReceive(pov->id, socket);
}

bool IOCPService::startReceive(ENDPOINT_ID id, SOCKET socket)
{
    DbgPrint("id=%ld, socket=%ld", id, socket);

    OV_DATA *pov;
    DWORD dwBytes, dwFlags = 0;

    pov = newOV_DATA();
    pov->type = OVT_RECEIVE;
    pov->id = id;
    pov->socket = socket;
    pov->packet.Set(NULL, TCP_PACKET_SIZE);

    if (WSARecv(pov->socket, &pov->packet(), 1, &dwBytes, &dwFlags, &pov->olp, NULL) != 0)
    {
        int err = WSAGetLastError();
        if (err != ERROR_IO_PENDING)
        {
            DbgPrint("WSARecv id=%ld failed, socket=%ld, err=%ld", pov->id, pov->socket, err);
            if (!postCompletion(pov->socket, 0, &pov->olp))
            {
                startClose(socket);
                deleteOV_DATA (pov);
            }
            return false;
        }
    }
    return true;
}

bool IOCPService::startClose(SOCKET socket)
{
    DbgPrint("id=%ld, socket=%ld", socket, socket);

    OV_DATA *pov = newOV_DATA();
    pov->type = OVT_CLOSE;
    pov->id = socket;
    pov->socket = socket;
    return postCompletion(socket, 0, &pov->olp);
}
void IOCPService::onClose(SOCKET socket, DWORD dwTransferred, OV_DATA* pov, int error)
{
    DbgPrint("id=%ld, socket=%ld", pov->id, socket);
    shutdown(pov->socket, SD_BOTH);
    closesocket(pov->socket);
    if (m_pPFEventHandler)
    {
        m_pPFEventHandler->onTcpClosed(this, pov->id);
    }
    {
        std::lock_guard<std::mutex> lock{m_ConnectLock};
        m_connectlist.erase(pov->id);
    }
}

STATUS IOCPService::setEventHandler(TCPEventHandler *pEventHandler)
{
    m_pPFEventHandler = pEventHandler;
    return STATUS_SUCCESS;
}

OV_DATA* IOCPService::newOV_DATA()
{
    OV_DATA* pov = new OV_DATA();
    std::lock_guard<std::mutex> lock{ m_ovDataLock };
    InsertTailList(&m_ovDataList, &pov->entry);
    m_ovDataCounter++;
    DbgPrint("ov set size=%ld", m_ovDataCounter);
    return pov;
}

void IOCPService::deleteOV_DATA(OV_DATA* pov)
{
    std::lock_guard<std::mutex> lock{ m_ovDataLock };
    RemoveEntryList(&pov->entry);
    InitializeListHead(&pov->entry);
    m_ovDataCounter--;
    DbgPrint("OV set size=%ld, id=%ld, type=%ld, socket=%ld", m_ovDataCounter, pov->id, pov->type, pov->socket);
    delete pov;
}